const { Transform } = require("stream");
const mongodb = require("mongodb");
const _ = require("lodash");
const utils = require('../../utils');

function getDB(uri) {
  return new Promise((resolve) => {
    const MongoClient = mongodb.MongoClient;
    MongoClient.connect(uri, { socketTimeoutMS: 10000000 }, function (err, db) {
      if (err) {
        utils.debug('mongodbWriter [error]:', err);
        process.exit(0);
      };
      resolve(db);
    });
  });
}

module.exports = async function ({ uri = 'mongodb://localhost:27017', database, table }) {
  const db = await getDB(uri);
  const stream = new Transform({ objectMode: true });

  stream._transform = async function (chunk, encoding, callback) {
    utils.debug('mongodbWriter [chunk]:', chunk);
    stream.push(chunk);
    const t = db.db(database).collection(table);
    _.isArray(chunk) ? await t.insertMany(chunk) : await t.insertOne(chunk);
    callback();
  };

  stream.on('finish', async function () {
    db.close();
    utils.debug('mongodbWriter [finish]');
  });

  return stream;
};
